/**
 * @file   queue_thread.h
 * @author Pedro Jimenez
 * @date   Fri Nov  7 13:44:03 2008
 * 
 * @brief  This class generate a threaded circular buffer to deal with the producer - consumer problem
 * 
 * 
 */

#ifndef _QUEUE_THREAD_H_
#define _QUEUE_THREAD_H_

#include <iostream>
using namespace std;


#include <sys/time.h>
#include <unistd.h>
#include <sys/select.h>

#include <queue>
#include <vector>


template <typename  T>
class queue_thread
{
 protected:
    std::queue <T*> buffers_use;
    std::queue <T*> buffers_free;
    T* buffer_in_use;
    pthread_mutex_t mutex_use;
    pthread_mutex_t mutex_free;
    
    pthread_t m_pthread;
    bool m_quit;
    int m_size;
    bool m_use_external_buffer;
 public:
    useconds_t m_sleep;
    T* circular_buffer;

 public:
    queue_thread(int size=20, T* buffer=NULL, useconds_t miliseconds=5000)
    {
	cout << "thread_queue()" << endl;

	m_size = size;
	m_quit = false;
	m_use_external_buffer = false;
	circular_buffer = 0;
	m_pthread = 0;
	//mutex_free=NULL;
	//mutex_use=NULL;
	m_sleep = miliseconds; //5 miliseconds default sleep time
	    
	if (buffer)
	{
	    m_use_external_buffer = true;
	    circular_buffer = buffer;
	}
	else //(buffer == 0)
	{
	    m_use_external_buffer = false;
	    circular_buffer = new T[m_size];
	}
	init();
    };
    ~queue_thread()
    {
	cout << "~thread_queue()" << endl;

	if (circular_buffer)
	{
	    deinit();
	    if (!m_use_external_buffer)
		delete[] circular_buffer;
	    circular_buffer = 0;
	}
    };

    virtual void init()
    {
	cout << "virtual void init()" << endl;
	if (circular_buffer)
	    for (int i=0; i<m_size; i++)
		buffers_free.push(circular_buffer+i);
	pthread_mutex_init(&mutex_free, NULL);
	pthread_mutex_init(&mutex_use, NULL);
    };
    virtual void deinit()
    {
	cout << "virtual void deinit()" << endl;
	while (!buffers_free.empty())
	    buffers_free.pop();
	while (!buffers_use.empty())
	    buffers_use.pop();
    };

    //(void * (void * params)) *pthread_funct;
    /*    {
	printf("queue_thread::pthread_funct\n");
	return NULL;
	};*/
    virtual bool start_thread()
    {
	cout << "virtual bool start_thread()" << endl;
	if (!m_pthread)
	{
	    //if (pthread_create(&m_pthread, NULL,&pthread_funct, (void*)this))
	    {
		return false;
		m_pthread =0;
	    }
	}
	return true;
    };
    
    virtual void push_free_buffer(T* g)
    {
	cout << "virtual void push_free_buffer()" << endl;
	pthread_mutex_lock(&mutex_free);
	buffers_free.push(buffer_in_use);
	pthread_mutex_unlock(&mutex_free);
    };

    virtual T* pop_used_buffer(void)
    {
	cout << "virtual T* pop_used_buffer()" << endl;
	pthread_mutex_lock(&mutex_use);
	while (!m_quit & buffers_use.empty())
	{
	    pthread_mutex_unlock(&mutex_use);
	    usleep(m_sleep);
	}
	if (m_quit)
	{
	    pthread_mutex_unlock(&mutex_use);
	    return (T*)(NULL);
	}
	buffer_in_use = buffers_use.front();
	buffers_use.pop();
	pthread_mutex_unlock(&mutex_use);
	return buffer_in_use;
    };

    // protected:
 public:
    virtual T* pop_free_buffer(void)
    { 
	cout << "virtual T* pop_free_buffer()" << endl;
	pthread_mutex_lock(&mutex_free);
	while (!m_quit & buffers_free.empty())
	{
	    pthread_mutex_unlock(&mutex_free);
	    usleep(m_sleep);
	}
	if (m_quit)
	{
	    pthread_mutex_unlock(&mutex_free);
	    return (T*)(0);
	}
	buffer_in_use = buffers_free.front();
	buffers_free.pop();
	pthread_mutex_unlock(&mutex_free);
	return buffer_in_use;
    };    
    virtual void push_use_buffer(T* g)
    {
	cout << "virtual void push_use_buffer()" << endl;
	pthread_mutex_lock(&mutex_use);
	buffers_use.push(buffer_in_use);
	pthread_mutex_unlock(&mutex_use);
    };
};

#endif
